package cool.taomu.toolkit.utils

import com.google.common.util.concurrent.FutureCallback
import com.google.common.util.concurrent.Futures
import com.google.common.util.concurrent.ListenableFuture
import com.google.common.util.concurrent.ListeningExecutorService
import com.google.common.util.concurrent.MoreExecutors
import java.util.ArrayList
import java.util.List
import java.util.concurrent.Callable
import java.util.concurrent.Executors
import java.util.concurrent.TimeUnit
import java.util.function.Function
import org.slf4j.LoggerFactory

@Deprecated
class ExecutorUtils {

    val static LOG = LoggerFactory.getLogger(ExecutorUtils)

    val static executor = MoreExecutors.listeningDecorator(
        Executors.newScheduledThreadPool(4)
    );

    /**
     * 提交多个 Callable 任务，并在所有任务完成后进行处理
     * 
     * @param <T> 任务返回值类型
     * @param executorService 执行器服务
     * @param tasks 一组 Callable 任务
     * @param fcall 可选的转换函数，用于处理最终结果
     * @param fcallback 可选的 Future 回调函数
     * @return ListenableFuture<List<T>> 可监听的结果 future
     */
    def static <T> ListenableFuture<List<T>> submit(
        ListeningExecutorService executorService,
        Iterable<Callable<T>> tasks,
        Function<List<T>, List<T>> fcall,
        FutureCallback<List<T>> fcallback
    ) {

        val futures = new ArrayList<ListenableFuture<T>>();

        for (task : tasks) {
            val future = executorService.submit(task)
            futures.add(future)
        }

        // 合并所有 future
        val allFutures = Futures.allAsList(futures)

        // 应用转换函数（如果存在）
        val transformedFuture = if (fcall !== null) {
                Futures.transform(allFutures, [ result |
                    LOG.info("执行转换函数")
                    fcall.apply(result)
                ], MoreExecutors.directExecutor)
            } else {
                allFutures
            }

        // 添加回调（如果存在）
        if (fcallback !== null) {
            LOG.info("执行回调函数")
            Futures.addCallback(transformedFuture, fcallback, MoreExecutors.directExecutor)
        }

        return transformedFuture
    }

    def static <T> ListenableFuture<List<T>> asyncSubmit(
        ListeningExecutorService executorService,
        Iterable<Callable<T>> tasks,
        Function<List<T>, List<T>> fcall,
        FutureCallback<List<T>> fcallback
    ) {

        val futures = new ArrayList<ListenableFuture<T>>();

        for (task : tasks) {
            val future = executorService.submit(task)
            futures.add(future)
        }

        // 合并所有 future
        val allFutures = Futures.allAsList(futures)

        // 应用转换函数（如果存在）
        val transformedFuture = if (fcall !== null) {
                Futures.transform(allFutures, [ result |
                    LOG.info("执行转换函数")
                    fcall.apply(result)
                ], executorService)
            } else {
                allFutures
            }

        // 添加回调（如果存在）
        if (fcallback !== null) {
            LOG.info("执行回调函数")
            Futures.addCallback(transformedFuture, fcallback, executorService)
        }

        return transformedFuture
    }

    /**
     * 提交多个 Callable 任务，并设置最大等待时间
     * 
     * @param <T> 任务返回值类型
     * @param executorService 执行器服务
     * @param tasks 一组 Callable 任务
     * @param timeout 超时时间
     * @param unit 时间单位
     * @param fcallback 可选的 Future 回调函数
     * @return ListenableFuture<List<T>> 带超时控制的 future
     */
    def static <T> ListenableFuture<List<T>> submitWithTimeout(
        ListeningExecutorService executorService,
        Iterable<Callable<T>> tasks,
        long timeout,
        TimeUnit unit,
        FutureCallback<List<T>> fcallback
    ) {

        val future = submit(executorService, tasks, null, null)

        // 设置超时限制
        val timeoutFuture = Futures.withTimeout(future, timeout, unit, executor)

        // 添加回调（如果存在）
        if (fcallback !== null) {
            LOG.info("执行回调函数")
            Futures.addCallback(timeoutFuture, fcallback, MoreExecutors.directExecutor)
        }

        return timeoutFuture
    }

    def static <T> ListenableFuture<List<T>> asyncSubmitWithTimeout(
        ListeningExecutorService executorService,
        Iterable<Callable<T>> tasks,
        long timeout,
        TimeUnit unit,
        FutureCallback<List<T>> fcallback
    ) {
        return asyncSubmitWithTimeout(executorService, tasks, timeout, unit, fcallback)
    }

    def static <T> ListenableFuture<List<T>> asyncSubmitWithTimeout(
        ListeningExecutorService executorService,
        Iterable<Callable<T>> tasks,
        Function<List<T>, List<T>> fcall,
        long timeout,
        TimeUnit unit,
        FutureCallback<List<T>> fcallback
    ) {

        val future = asyncSubmit(executorService, tasks, fcall, null)

        // 设置超时限制
        val timeoutFuture = Futures.withTimeout(future, timeout, unit, executor)

        // 添加回调（如果存在）
        if (fcallback !== null) {
            LOG.info("执行回调函数")
            Futures.addCallback(timeoutFuture, fcallback, executorService)
        }

        return timeoutFuture
    }
}
